package com.nami.dispatcher.domain.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import com.google.protobuf.ByteString;
import com.nami.dispatcher.domain.message.model.PacketTrans;
import com.nami.dispatcher.domain.middleware.mq.DomainEventPublisher;
import com.nami.dispatcher.domain.model.NtfSenderModel;
import com.nami.dispatcher.domain.qos.QosSendDaemonS2C;
import com.nami.dispatcher.domain.service.DispatchCommonDomainService;
import com.nami.gate.external.basic.cache.user.DeviceInfo;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.im.network.model.enums.DownMsgErrorCode;
import com.nami.im.network.model.enums.protocol.CMDEnum;
import com.nami.im.network.model.message.v2.Message;
import com.nami.transport.proto.codec.DownWsMsg;
import com.nami.transport.proto.codec.ImMessages;
import com.nami.transport.proto.codec.MqTransMsg;
import com.nami.transport.proto.codec.UpWsMsg;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.utils.SessionUtil;
import org.dromara.common.json.utils.JsonUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

import static com.nami.gate.external.basic.constants.CommonConst.COLON;


@Slf4j
@Service
@RequiredArgsConstructor
public class DispatchCommonDomainServiceImpl implements DispatchCommonDomainService {

    private final DomainEventPublisher mqSender;

    @Override
    public void failResp(Message downMsg,
                         String sessionId,
                         String connectId, String code) {
        DownWsMsg.PublishAckMsgBody ack = DownWsMsg.PublishAckMsgBody.newBuilder()
            .setClientMsgId(downMsg.getClientUid())
            .setMsgId(downMsg.getMsgId())
            .setTimestamp(System.currentTimeMillis())
            .setMsgSeqNo(downMsg.getMsgSeqNo())
            .setCode(code)
            .build();

        MqTransMsg.MessageHeader header = MqTransMsg.MessageHeader.newBuilder()
            .setCmd(ByteString.copyFrom(new byte[]{CMDEnum.PUBACK.getCode()}))
            .setQos(ByteString.copyFrom(new byte[]{0}))
            .setVersion(ByteString.copyFrom(new byte[]{0}))
            .build();
        MqTransMsg.Dispatch2GatePublishResponse response = MqTransMsg.Dispatch2GatePublishResponse.newBuilder()
            .setSessionId(sessionId)
            .setMsgHeader(header)
            .setDownMsg(ack.toByteString())
            .build();
        mqSender.publish(DownTopicConst.FLUSH_ONE_TOPIC, connectId, response.toByteArray());
    }


    @Override
    public void notifySender(NtfSenderModel downTransModel) {
        // 转发我自己
        replyAck2CurrDevice(downTransModel);
        //
        syncMsg2SenderClients(downTransModel);
    }

    /**
     * 回复发送消息的设备， ack
     * 系统已收到消息。客户端可以停止qos了
     */
    private void replyAck2CurrDevice(final NtfSenderModel notify) {
        Message message = notify.getMessage();
        DownWsMsg.PublishAckMsgBody ack = DownWsMsg.PublishAckMsgBody.newBuilder()
            .setClientMsgId(message.getClientUid())
            .setMsgId(message.getMsgId())
            .setTimestamp(System.currentTimeMillis())
            .setMsgSeqNo(message.getMsgSeqNo())
            .setCode(DownMsgErrorCode.SUCCESS.getCode())
            .build();
        MqTransMsg.MessageHeader header = MqTransMsg.MessageHeader.newBuilder()
            .setCmd(ByteString.copyFrom(new byte[]{CMDEnum.PUBACK.getCode()}))
            .setQos(ByteString.copyFrom(new byte[]{0}))
            .setVersion(ByteString.copyFrom(new byte[]{0}))
            .build();
        MqTransMsg.Dispatch2GatePublishResponse response = MqTransMsg.Dispatch2GatePublishResponse.newBuilder()
            .setMsgHeader(header)
            .setSessionId(notify.getSessionId())
            .setMsgId(message.getMsgId())
            .setTimestamp(System.currentTimeMillis())
            .setDownMsg(ack.toByteString())
            .build();
        mqSender.publish(DownTopicConst.FLUSH_ONE_TOPIC, notify.getTag(), response.toByteArray());
        log.info("[DISPATCHER] 转发给发送设备ack完毕。sessionId={}", notify.getSessionId());
    }

    /**
     * 生成消息指纹
     *
     * @param userNo
     * @param clientMsgId
     * @return
     */
    @NotNull
    protected static String genUserMsgFingerChar(String userNo, String clientMsgId) {
        return userNo + COLON + clientMsgId;
    }

    private ImMessages.DownMsg toDownMsg(Message message) {
        ImMessages.DownMsg downMsg = ImMessages.DownMsg.newBuilder()
            .setClientUid(message.getClientUid())
            .setMsgId(message.getMsgId())
            .setMsgContent(ByteString.copyFromUtf8(message.getMsgContent()))
            .setMsgTime(message.getMsgTime())
            .setPlatform(message.getPlatform())
            .setChannelType(ImMessages.ChannelType.Private)
            .setSenderId(message.getSenderId())
            .setTargetId(message.getTargetId())
            .setMsgSeqNo(message.getMsgSeqNo())
            .setMsgType(message.getMsgType())
            .build();
        return downMsg;
    }

    private void syncMsg2SenderClients(NtfSenderModel sender) {
        Message message = sender.getMessage();
        ImMessages.DownMsg downMsg = toDownMsg(message);

        UpWsMsg.PublishMsg publishMsg = UpWsMsg.PublishMsg.newBuilder()
            .setTopic("sync_msg")
            .setData(downMsg.toByteString())
            .setTargetId(message.getSenderId())
            .build();
        // 向from其他在线客户端同步消息
        if (CollectionUtil.isNotEmpty(sender.getOtherClients())) {
            for (DeviceInfo client : sender.getOtherClients()) {
                log.info("[DISPATCHER] 消息同步 client:{}", (client.getPlatform()));
                MqTransMsg.MessageHeader header = MqTransMsg.MessageHeader.newBuilder()
                    .setCmd(ByteString.copyFrom(new byte[]{CMDEnum.PUBLISH.getCode()}))
                    .setQos(ByteString.copyFrom(new byte[]{0}))
                    .setVersion(ByteString.copyFrom(new byte[]{0}))
                    .build();
                MqTransMsg.Dispatch2GatePublishResponse response = MqTransMsg.Dispatch2GatePublishResponse.newBuilder()
                    .setMsgHeader(header)
                    .setSessionId(SessionUtil.binding(message.getSenderId(), client.getInstanceId()))
                    .setMsgId(message.getMsgId())
                    .setTimestamp(System.currentTimeMillis())
                    .setDownMsg(publishMsg.toByteString())
                    .build();

                //ForwardPacketUtil.flushPacket(request, client.getDubboIp(), client.getDubboPort());
                mqSender.publish(DownTopicConst.FLUSH_ONE_TOPIC, sender.getTag(), response.toByteArray());
                log.info("[DISPATCHER] 转发发送者其他客户端。sessionId={}", response.getSessionId());

                QosSendDaemonS2C.getInstance()
                    .put(genQosTransBo(response.toByteArray(), response.getSessionId(), message.getMsgId(),
                        sender.getAppKey(), client.getConnectId()));
            }

        }
    }

    @NotNull
    protected PacketTrans genQosTransBo(byte[] packet, String sessionId, String msgId,
                                        String appId, String mqTags) {
        PacketTrans packetTransBO = new PacketTrans();
        packetTransBO.setPacket(packet);
        packetTransBO.setQos(true);
        packetTransBO.setSessionId(sessionId);
        packetTransBO.setMessageId(msgId);
        packetTransBO.setAppKey(appId);

        packetTransBO.setMqTags(mqTags);

        packetTransBO.setIfGroupMsg(true);
        return packetTransBO;
    }

}
